package spark;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.Arrays;
import java.util.List;
import java.util.StringJoiner;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.apache.log4j.Logger;

public class ShellBridge {
    static final Logger LOGGER = Logger.getLogger(ShellBridge.class);


    public static void main(String[] args) throws IOException, InterruptedException {


        String cmd = "/usr/local/spark/bin/spark-submit --class org.apache.spark.examples.SparkPi --master spark://master:7077 /usr/local/spark/examples/jars/spark-examples*.jar 10";
        cmd = "/usr/local/spark/bin/spark-submit --class org.apache.spark.examples.SparkPi --master yarn --deploy-mode client /usr/local/spark/examples/jars/spark-examples*.jar 10";
        if (args.length > 0){
            StringJoiner joiner = new StringJoiner(" ");
            for (String s:args) {
                joiner.add(s);
            }
            cmd = joiner.toString();
        }

        List<String> cmds = Arrays.asList("sh", "-c", cmd);
        ProcessBuilder processBuilder = new ProcessBuilder(cmds);

        processBuilder.redirectErrorStream(true);
        final Process proc = processBuilder.start();

        new Thread(new Runnable() {
            public void run () {
                try {
                    BufferedReader outbufferedReader = new BufferedReader(new InputStreamReader(proc.getInputStream(),"GBK"));
                    String line = null;

                    while ((line = outbufferedReader.readLine()) != null) {
                        LOGGER.info(line);

                        String pattern = "Application report for (.+) \\(state:( [A-Z]+)\\)";

                        // 创建 Pattern 对象
                        Pattern r = Pattern.compile(pattern);

                        // 现在创建 matcher 对象
                        Matcher m = r.matcher(line);
                        if (m.find( )) {
                            System.out.println("Application: " + m.group(1) );
                            System.out.println("state: " + m.group(2) );
                        }
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }).start();

        System.out.println("Waiting for finish...");
        int exitCode = proc.waitFor();
        System.out.println("Finished! Exit code:" + exitCode);

    }
}
